package com.guanzi.big_data_mq_process.schedule;

import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.guanzi.big_data_mq_process.constant.MessageConstant;
import com.guanzi.big_data_mq_process.model.OrderCreateContext;
import com.guanzi.big_data_mq_process.model.dto.OrdersDTO;
import com.guanzi.big_data_mq_process.model.entity.TransactionMessage;
import com.guanzi.big_data_mq_process.model.enums.LocalTransactionStateEnum;
import com.guanzi.big_data_mq_process.service.ITransactionMessageService;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.stream.Collectors;

@Component
public class MessageRecoveryTask {

    @Autowired
    private ITransactionMessageService transactionMessageService;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    // 定时兜底任务，1分钟捞一次未提交的事务消息
//    @Scheduled(fixedRate = 60 * 1000)
    public void recoveryMessages() {
        List<TransactionMessage> messageList = transactionMessageService.list(
                Wrappers.<TransactionMessage>query().in("tx_state", LocalTransactionStateEnum.PREPARED, LocalTransactionStateEnum.ROLLBACKED));

        List<TransactionMessage> prepareMessageList = messageList.stream().filter(item -> LocalTransactionStateEnum.PREPARED.equals(item.getTxState())).collect(Collectors.toList());
        List<TransactionMessage> rollbackMessageList = messageList.stream().filter(item -> LocalTransactionStateEnum.ROLLBACKED.equals(item.getTxState())).collect(Collectors.toList());

        // 处理未提交的事务消息
        if (CollectionUtils.isNotEmpty(prepareMessageList)) {
            for (TransactionMessage localMsg : prepareMessageList) {
                // TODO 为什么是sendOneWay
                rocketMQTemplate.sendOneWay(localMsg.getTopic() + ":" + localMsg.getTag(),
                        MessageBuilder.withPayload(JSON.parseObject(localMsg.getPayload(), OrderCreateContext.class))
                                .setHeader(RocketMQHeaders.TRANSACTION_ID, localMsg.getMessageId())
                                .build()
                        );
            }
        }

        // 处理失败的事务消息
        if (CollectionUtils.isNotEmpty(rollbackMessageList)) {
            for (TransactionMessage localMsg : rollbackMessageList) {
                // TODO 为什么是syncSendDelayTimeSeconds
                rocketMQTemplate.syncSendDelayTimeSeconds(
                        localMsg.getTopic() + ":RETRY",
                        MessageBuilder.withPayload(JSON.parseObject(localMsg.getPayload(), OrderCreateContext.class))
                                .setHeader(RocketMQHeaders.TRANSACTION_ID, localMsg.getMessageId())
                                .build(),
                            calculateDelay(localMsg.getRetryCount())
                        );
            }
        }

    }

    /**
     * 计算 延迟时间
     *
     * @param retryCount 重试次数
     * @return
     */
    private int calculateDelay(int retryCount) {
        return switch (retryCount) {
            case 0 -> 300;  // 5分钟
            case 1 -> 600;  // 10分钟
            default -> 1800; // 30分钟
        };
    }

}
